package com.suning.sawp.service.impl.bi.support;

import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

import com.google.common.eventbus.AsyncEventBus;
import com.google.common.eventbus.EventBus;
import com.suning.istore.common.integration.ListeningAction;

public class BIAsyncListener implements InitializingBean, DisposableBean, ListeningAction.Listener<Object> {

    private EventBus asyncEventBus;

    private ThreadPoolExecutor executor;

    @Override
    public void afterPropertiesSet() throws Exception {
        executor = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,
                new SynchronousQueue<Runnable>());
        asyncEventBus = new AsyncEventBus(executor);

    }

    public void register(Object object) {
        asyncEventBus.register(object);
    }

    @Override
    public void complete(Object[] objs) {
        for (Object event : objs) {
            asyncEventBus.post(event);
        }
    }

    @Override
    public void destroy() throws Exception {
        executor.shutdownNow();
    }

}
